package schedule

import (
	"container/list"
	"fmt"
	"mapreduce-go/call"
	"mapreduce-go/common"
	"sync"
)

func Schedule(
	jobName string,
	mapFiles []string,
	nReduce int,
	phase common.JobPhase,
	registerChan chan *string) {

	// 当前任务数量
	var nTasks int

	// 另一个任务数量
	var nOther int

	switch phase {
	case common.MapPhase:
		nTasks = len(mapFiles)
		nOther = nReduce
	case common.ReducePhase:
		nTasks = nReduce
		nOther = len(mapFiles)
	}

	// 执行mapreduce的函数功能调用操作
	// 所有任务必须调度给workers 在等到所有任务全部执行成功后 在下一步处理
	// 一个worker可以完成多个任务

	var wg sync.WaitGroup
	var lock = &sync.Mutex{}

	wg.Add(nTasks)

	// 声明一个任务列表 将所有待处理的任务添加进去
	taskList := list.New()
	for i := 0; i < nTasks; i++ {
		taskList.PushBack(i)
	}

	for {

		lock.Lock()
		if taskList.Len() <= 0 {
			lock.Unlock()
			break
		}

		elm := taskList.Front()
		taskId := elm.Value.(int)
		taskList.Remove(elm)

		lock.Unlock()

		taskArgs := &call.DoTaskArgs{
			TaskId:           taskId,
			Phase:            phase,
			JobName:          jobName,
			NumberOtherPhase: nOther,
		}

		if phase == common.MapPhase {
			taskArgs.File = mapFiles[taskId]
		}

		worker := <-registerChan
		fmt.Println("master : schedule : ", *worker)
		registerChan <- worker

		// 调用rpc
		go func() {
			fmt.Println("master : call do task, worker : ", *worker, " args : ", taskArgs)
			ok := call.Call(*worker, "Worker.DoTask", taskArgs, nil)

			if ok {
				fmt.Println("master : call ok")
				wg.Done()
			} else {
				fmt.Println("master : task failed")
				// worker执行任务失败
				// 将该任务重新加入列表 相当于重新入队
				lock.Lock()
				taskList.PushBack(taskArgs.TaskId)
				lock.Unlock()
				wg.Done()
			}
		}()
	}

	wg.Wait()
}
